package com.lsx143.realtime.app.dws;

import com.lsx143.realtime.app.BaseSQLApp;
import com.lsx143.realtime.common.Constants;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DWSKeyWordStatsApp extends BaseSQLApp {
    private final static String appName = "DWSKeyWordStatsApp";

    public static void main(String[] args) {
        new DWSKeyWordStatsApp().init(
                30005,
                appName,
                1
        );
    }

    @Override
    protected void run(StreamTableEnvironment tEnv) {
        // 1. 建一张动态表与topic关联
        tEnv.executeSql("create table page_log(" +
                "   page map<string, string>, " +
                "   ts bigint, " +
                "   et as to_timestamp(from_unixtime(ts/1000)), " +
                "   watermark for et as et - interval '3' second " +
                ")with(" +
                "   'connector' = 'kafka'," +
                "   'properties.bootstrap.servers' = '" + Constants.KAFKA_BROKERS + "'," +
                "   'properties.group.id' = '" + appName + "'," +
                "   'topic' = '" + Constants.TOPIC_DWD_PAGE_LOG + "'," +
                "   'scan.startup.mode' = 'latest-offset', " +
                "   'format' = 'json' " +
                ")");

        tEnv.executeSql("select * from page_log limit 10").print();
    }
}
